package com.wyt.mq.mqtt;

import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
public class MqttAsyncClientUtils {

    private static Logger logger = LoggerFactory.getLogger(MqttAsyncClientUtils.class);

    public static Map<String, MqttAsyncClient> clientIdMqttMap = new HashMap<>();


    /********************** 静态变量赋值 ***************************/
    public static String host;
    @Value("${vehicle.mqtt.host}")
    public void setHost(String host) {
        MqttClientUtils.host = host;
    }
    public static String username;
    @Value("${vehicle.mqtt.username}")
    public void setUsername(String username) {
        MqttClientUtils.username = username;
    }
    public static String password;
    @Value("${vehicle.mqtt.password}")
    public void setPassword(String password) {
        MqttClientUtils.password = password;
    }

    /********************** 静态变量赋值 ***************************/

    /**
     * 获取客户端
     */
    public static MqttAsyncClient getClient(Class callback, String clientId) throws Exception {
        if (!clientIdMqttMap.containsKey(clientId)) {
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setCleanSession(true);
            options.setAutomaticReconnect(true);
            options.setConnectionTimeout(320);
            MqttAsyncClient client = new MqttAsyncClient(host, clientId, new MemoryPersistence());
            //@TODO 失效重连的方法中，要用reconnect方法测试，而且重连后要重新订阅topic
            if (callback != null) client.setCallback((MqttCallbackExtended) callback.newInstance());
            client.connect(options);
            clientIdMqttMap.put(clientId, client);
        }
        MqttAsyncClient client = clientIdMqttMap.get(clientId);
        if (!client.isConnected()) client.connect();
        return client;
    }

    public static MqttAsyncClient getClient(MqttCallbackExtended callback, String clientId) throws Exception {
        if (!clientIdMqttMap.containsKey(clientId)) {
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setCleanSession(true);
            options.setAutomaticReconnect(true);
            options.setConnectionTimeout(320);
            MqttAsyncClient client = new MqttAsyncClient(host, clientId, new MemoryPersistence());
            //@TODO 失效重连的方法中，要用reconnect方法测试，而且重连后要重新订阅topic
            if (callback != null) client.setCallback(callback);
            client.connect(options);
            clientIdMqttMap.put(clientId, client);
        }
        MqttAsyncClient client = clientIdMqttMap.get(clientId);
        if (!client.isConnected()) client.connect();
        return client;
    }

    /**
     * 断开重连
     */
    public static void reconnect(String clientId) {
        MqttAsyncClient client = clientIdMqttMap.get(clientId);
        int cnt = 1;
        while (client != null && !client.isConnected()) {
            try {
                logger.info("{}第{}次重连mqtt", clientId, cnt++);
                client.reconnect();
                Thread.sleep(1000);
            } catch (Exception e) {
                logger.error("{} mqtt重连失败", clientId);
            }
        }
    }

    /**
     * 推送消息
     */
    public static void publish(String pushMessage, String topic, String clientId) {
        MqttMessage message = new MqttMessage();
        message.setPayload(pushMessage.getBytes());
        try {
            MqttAsyncClient client = getClient((Class) null, clientId);
            client.publish(topic, message);
        } catch (Exception e) {
            logger.error("{}向{} topic发送数据异常{}", clientId, topic, e.getMessage());
            throw new RuntimeException(e);
        }
    }







}
